package com.xwj.SparkOperator

import com.xwj.myclass.MyConnection
import org.apache.spark.SparkContext

/**
  * Created by thinkjoy on 2017/8/17.
  * 常用rdd算子
  */
object spark_operator {

  var spark:SparkContext=null


  def main(args: Array[String]): Unit = {
//  map_operator
//  flatMap_operator
  mapPartitions_operator
//  sample_operator
//    union_operator
//    intersection_operator
//    distinct_operator
//    cartesian_operator
//    glom_operator
//    randomSplit_operator
  }

  /**
    * map算子
    */
  def map_operator(): Unit ={
    spark=new MyConnection().connect()
    val rdd=spark.parallelize(1 to 10)  //创建rdd
    rdd.map(_*2).collect.foreach(println)  //每个元素乘2，，集合化，遍历输出
    spark.stop()
  }

  /**
    * flatmap算子
    * 将原来 RDD 中的每个元素通过函数 f 转换为新的元素，
    * 并将生成的 RDD 的每个集合中的元素合并为一个集合
    */
  def flatMap_operator():Unit={
    spark=new MyConnection().connect()
    spark.parallelize(1 to 10).
      flatMap(x=>(1 to x)).collect.foreach(x=>print(x+" "))
    spark.stop()
  }

  /**
    * mapPartitions算子
    *  获 取 到 每 个 分 区 的 迭 代器，在 函 数 中 通 过 这 个 分 区 整 体 的 迭 代 器
    *  对整 个 分 区 的 元 素 进 行 操 作
    *  _2：二元组的的第二个元素
    *  _._2等价t=>t._2
    */
  def mapPartitions_operator: Unit={
    spark=new MyConnection().connect()
    val lits=List(("kpop","female"),("zorro","male"),("mobin","male"),("lucy","female"))
    spark.parallelize(lits,2).mapPartitions(x=>x.filter(_._2=="female")).
      map(x=>x._1).collect().foreach(println)
    spark.stop()
  }

  /**
    * sample(withReplacement,fraction,seed):以指定的随机种子随机抽样出数量为fraction的数据，
    * withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样
    * seed为随机种子值
    */
  def sample_operator:Unit={
    spark=new MyConnection().connect()
    spark.parallelize(1 to 10).sample(true,0.5,3).
      collect().foreach(x=>print(x+" "))
    spark.stop()
  }

  /**
    * union(ortherDataset):将两个RDD中的数据集进行合并，最终返回两个RDD的并集，
    * 若RDD中存在相同的元素也不会去重
    */
  def union_operator:Unit={
    spark=new MyConnection().connect()
    val rdd1=spark.parallelize(1 to 3)
    val rdd2=spark.parallelize(3 to 5)
    rdd1.union(rdd2).collect().foreach(println)
    spark.stop()
  }

  /**
    * intersection(otherDataset):返回两个RDD的交集
    */
  def intersection_operator:Unit={
    spark=new MyConnection().connect()
    val rdd1=spark.parallelize(1 to 3)
    val rdd2=spark.parallelize(3 to 5)
    rdd1.intersection(rdd2).collect().foreach(println)
    spark.stop()
  }

  /**
    * distinct([numTasks]):对RDD中的元素进行去重
    */
  def distinct_operator:Unit={
    spark=new MyConnection().connect()
    val list=List(1,1,2,5,2,9,6,1)
    spark.parallelize(list).distinct().collect().foreach(println)
    spark.stop()
  }

  /**
    * cartesian(otherDataset):对两个RDD中的所有元素进行笛卡尔积操作
    */
  def cartesian_operator:Unit={
    spark=new MyConnection().connect()
    val rdd1 = spark.parallelize(1 to 3)
    val rdd2 = spark.parallelize(2 to 5)
    rdd1.cartesian(rdd2).collect().foreach(println)
    spark.stop()
  }

  /**
    * glom():将RDD的每个分区中的类型为T的元素转换换数组Array[T]
    */
  def glom_operator:Unit={
    spark=new MyConnection().connect()
    spark.parallelize(1 to 16,4).glom().collect().
      foreach(rdd=>println(rdd.getClass.getSimpleName))
    spark.stop()
  }

  /**
    * randomSplit(weight:Array[Double],seed):根据weight权重值将一个RDD划分成多个RDD,
    * 权重越高划分得到的元素较多的几率就越大
    */
  def randomSplit_operator:Unit={
    spark=new MyConnection().connect()
    val randomSplitRDD =spark.parallelize(1 to 10).randomSplit(Array(1.0,2.0,7.0))
    randomSplitRDD(0).collect().foreach(x=>print(x+" "))
    randomSplitRDD(1).collect().foreach(x=>print(x+" "))
    randomSplitRDD(2).collect().foreach(x=>print(x+" "))
    spark.stop()
  }

}
